package com.sh.data.engine.domain.datadev.flink.service.impl;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkNodeConfigDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkResourceDomain;
import com.sh.data.engine.domain.datadev.flink.model.param.SaveResourceParam;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeConfigService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkResourceService;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkMenuNodeEntity;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkNodeConfigEntity;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.IOException;
import java.util.*;

/**
 * @author：jingchen.hr
 * @descripion:
 * @data:6/16/20221:51 PM
 */
@Service
public class FlinkRrsourceServiceImpl implements FlinkResourceService {

    @Lazy
    @Autowired
    private FlinkNodeConfigService flinkNodeConfigService;
    @Lazy
    @Autowired
    private FlinkNodeService flinkNodeService;

    @Value("${flink.resource.path:/data}")
    private String flinkResourcePath;

    @Override
    public void delete(Long nodeId) {
        if (null == nodeId) {
            throw new BusinessException("该文件不存在，请重新刷新");
        }
        FlinkResourceDomain activeByNodeId = this.getActiveByNodeId(nodeId);

        FlinkNodeConfigEntity flinkResourceEntity =
            flinkNodeConfigService.getById(activeByNodeId.getId());
        if (flinkResourceEntity.getIsPublished().equals(1)) {
            throw new BusinessException("该jar包已经发布,不能删除");
        }
        flinkNodeConfigService.deleteResourceById(flinkResourceEntity.getId());
    }

    @Override
    public void deleteById(Long id) {
        FlinkNodeConfigEntity flinkResourceEntity = flinkNodeConfigService.getById(id);
        if (flinkResourceEntity.getIsPublished().equals(1)) {
            throw new BusinessException("该jar包已经发布,不能删除");
        }
        flinkNodeConfigService.removeById(id);
    }

    @Override
    public FlinkResourceDomain getActiveByNodeId(Long nodeId) {
        if (null == nodeId) {
            return null;
        }
        FlinkNodeConfigEntity flinkResourceEntity = flinkNodeConfigService.getActiveByNodeId(nodeId);

        if (null == flinkResourceEntity) {
            throw new BusinessException("不存在最新版本的资源节点");
        }
        FlinkResourceDomain flinkResourceDomain = new FlinkResourceDomain();
        flinkResourceDomain.init(flinkResourceEntity);
        return flinkResourceDomain;
    }

    @Override
    public List<FlinkResourceDomain> getActiveListByNodeIds(List<Long> nodeIds) {
        if (CollectionUtils.isEmpty(nodeIds)) {
            return Collections.EMPTY_LIST;
        }
        List<FlinkNodeConfigEntity> byNodeIds = flinkNodeConfigService.getByNodeIds(nodeIds);
        List<FlinkResourceDomain> list = Lists.newArrayList();
        byNodeIds.forEach(
            i -> {
                FlinkResourceDomain flinkResourceDomain = new FlinkResourceDomain();
                flinkResourceDomain.init(i);
                list.add(flinkResourceDomain);
            });
        return list;
    }

    @Override
    public FlinkResourceDomain getById(Long id) {
        FlinkNodeConfigDomain nodeConfigById = flinkNodeConfigService.getNodeConfigById(id);
        FlinkNodeConfigEntity entity =
            ConvertUtil.copyProperties(nodeConfigById, FlinkNodeConfigEntity.class);
        FlinkResourceDomain flinkResourceDomain = new FlinkResourceDomain();
        flinkResourceDomain.init(entity);
        return flinkResourceDomain;
    }

    @Override
    public void publish(Long nodeId, Integer status, String userId) {
        final FlinkResourceDomain flinkFunctionDomain = this.getActiveByNodeId(nodeId);
        if (null == flinkFunctionDomain) {
            throw new BusinessException("数据不存在");
        }

        if (Objects.equals(flinkFunctionDomain.getIsPublished(), 1)) {
            throw new BusinessException("已发布，不能重复发布");
        }

        FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
        flinkNodeConfigEntity.setId(flinkFunctionDomain.getId());
        flinkNodeConfigEntity.setIsPublished(1);

        if (Objects.equals(status, 1)) {
            flinkNodeConfigEntity.setPublishTime(new Date());
            flinkNodeConfigEntity.setPublisherId(userId);
        } else {
            flinkNodeConfigEntity.setPublishTime(null);
            flinkNodeConfigEntity.setPublisherId(null);
        }

        UpdateWrapper<FlinkNodeConfigEntity> wrapper = Wrappers.update();
        wrapper
            .lambda()
            .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
            .eq(FlinkNodeConfigEntity::getIsActive, 1);

        flinkNodeConfigService.update(flinkNodeConfigEntity, wrapper);
    }

    @Override
    public void rollbackVersion(Long nodeId, Long versionId) {

        UpdateWrapper<FlinkNodeConfigEntity> wrapper = Wrappers.update();
        wrapper
            .lambda()
            .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
            .eq(FlinkNodeConfigEntity::getIsActive, 1);
        FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
        flinkNodeConfigEntity.setIsActive(1);
        flinkNodeConfigEntity.setUpdateTime(new Date());
        flinkNodeConfigEntity.setId(versionId);
        flinkNodeConfigService.update(flinkNodeConfigEntity, wrapper);
    }

    @Override
    public Long uploadFlinkResource(Long projectId, String fileName, byte[] data) {
        String newFileName = UUID.randomUUID().toString();
        newFileName = String.format("%s.jar", newFileName);
        String path =
            String.format(
                "%s%s%s%s%s",
                flinkResourcePath, File.separator, projectId, File.separator, newFileName);
        String userId = UserContext.getUserId();
        FlinkNodeConfigEntity entity = new FlinkNodeConfigEntity();
        FlinkNodeConfigEntity.ResourceConfig resourceConfig =
            new FlinkNodeConfigEntity.ResourceConfig();

        File file = new File(path);
        try {
            FileUtils.writeByteArrayToFile(file, data);
        } catch (IOException e) {
            throw new BusinessException("上传失败");
        }
        entity.setCreateTime(new Date());
        resourceConfig.setFileName(fileName);
        entity.setUpdaterId(userId);
        resourceConfig.setPath(path);
        entity.setProjectId(projectId);
        entity.setUpdateTime(new Date());
        entity.setResourceConfig(resourceConfig);
        flinkNodeConfigService.save(entity);
        return entity.getId();
    }

    @Override
    public boolean saveFlinkResource(SaveResourceParam saveResourceParam) {
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        if (saveResourceParam.getId() == null) {
            throw new BusinessException("文件上传失败，请重新上传或者等待文件上传成功再保存");
        }

        FlinkNodeConfigEntity flinkResourceEntity =
            flinkNodeConfigService.getById(saveResourceParam.getId());
        FlinkNodeConfigEntity.ResourceConfig resourceConfig =
            new FlinkNodeConfigEntity.ResourceConfig();
        /** 不存在nodeId 新建，存在替换 */
        if (null == saveResourceParam.getNodeId()) {
            FlinkMenuNodeEntity flinkMenuNodeEntity = new FlinkMenuNodeEntity();
            flinkMenuNodeEntity.setCreateTime(new Date());
            flinkMenuNodeEntity.setIsLocked(0);
            flinkMenuNodeEntity.setIsSystem(0);
            flinkMenuNodeEntity.setUpdaterId(userId);
            flinkMenuNodeEntity.setPid(saveResourceParam.getPid());
            flinkMenuNodeEntity.setName(saveResourceParam.getResourceName());
            flinkMenuNodeEntity.setCreatorId(userId);
            flinkMenuNodeEntity.setProjectId(projectId);
            flinkMenuNodeEntity.setNodeType(FlinkNodeTypeEnum.RESOURCE.getCode());
            flinkNodeService.save(flinkMenuNodeEntity);
            // fixme 验证
            flinkResourceEntity.setNodeId(flinkMenuNodeEntity.getId());
        } else {
            /** 获取到最新的 资源节点 */
            FlinkResourceDomain activeNode = this.getActiveByNodeId(saveResourceParam.getNodeId());

            FlinkNodeConfigEntity resourceEntity =
                ConvertUtil.copyProperties(activeNode, FlinkNodeConfigEntity.class);
            resourceEntity.setIsActive(0);
            resourceEntity.setUpdateTime(new Date());
            resourceEntity.setUpdaterId(userId);
            flinkNodeConfigService.updateById(resourceEntity);
            flinkResourceEntity.setUpdaterId(userId);
            flinkResourceEntity.setCreateTime(new Date());
            flinkResourceEntity.setUpdateTime(new Date());
            flinkResourceEntity.setNodeId(activeNode.getNodeId());
        }
        flinkResourceEntity.setIsActive(1);
        resourceConfig.setResourceName(saveResourceParam.getResourceName());
        if (null != saveResourceParam.getRemark()) {
            resourceConfig.setRemark(saveResourceParam.getRemark());
        }
        flinkNodeConfigService.updateById(flinkResourceEntity);
        return true;
    }

    @Override
    public FlinkResourceDomain getFlinkResourceByNodeId(Long id) {
        FlinkNodeConfigEntity flinkResourceEntity = flinkNodeConfigService.getActiveByNodeId(id);
        FlinkResourceDomain flinkResourceDomain = new FlinkResourceDomain();
        flinkResourceDomain.init(flinkResourceEntity);
        return flinkResourceDomain;
    }

    @Override
    public IPage<FlinkResourceDomain> getFlinkResourceList(
        Long nodeId, Integer pageNum, Integer pageSize) {
        LambdaQueryWrapper<FlinkNodeConfigEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
                .orderByDesc(FlinkNodeConfigEntity::getCreateTime);

        IPage<FlinkResourceDomain> page =
            flinkNodeConfigService
                .page(new Page<>(pageNum, pageSize), lambdaQueryWrapper)
                .convert(e -> Convert.convert(FlinkResourceDomain.class, e));

        return page;
    }

    @Override
    public Boolean editFlinkResourceByNodeId(Long id, String resourceName) {
        String userId = UserContext.getUserId();

        FlinkMenuNodeEntity flinkMenuNodeEntity = flinkNodeService.getById(id);
        flinkMenuNodeEntity.setName(resourceName);
        flinkMenuNodeEntity.setUpdateTime(new Date());
        flinkNodeService.updateById(flinkMenuNodeEntity);

        FlinkResourceDomain activeByNodeId = this.getActiveByNodeId(id);
        FlinkNodeConfigEntity flinkResourceEntity =
            ConvertUtil.copyProperties(activeByNodeId, FlinkNodeConfigEntity.class);
        FlinkNodeConfigEntity.ResourceConfig resourceConfig =
            new FlinkNodeConfigEntity.ResourceConfig();
        flinkResourceEntity.setUpdaterId(userId);
        flinkResourceEntity.setUpdateTime(new Date());
        resourceConfig.setResourceName(resourceName);
        flinkNodeConfigService.updateById(flinkResourceEntity);
        return true;
    }

    @Override
    public List<FlinkResourceDomain> getFlinkResourceList(Long projectId) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> eq =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getProjectId, projectId)
                .eq(FlinkNodeConfigEntity::getIsActive, 1);
        List<FlinkNodeConfigEntity> multi = flinkNodeConfigService.list(eq);
        List<FlinkResourceDomain> flinkResourceDomains =
            ConvertUtil.copyProperties(multi, FlinkResourceDomain.class);
        return flinkResourceDomains;
    }

    @Override
    public FlinkResourceDomain getFlinkResourceById(Long id) {
        FlinkNodeConfigEntity byId = flinkNodeConfigService.getById(id);
        FlinkResourceDomain flinkResourceDomain =
            ConvertUtil.copyProperties(byId, FlinkResourceDomain.class);
        return flinkResourceDomain;
    }
}
